package cn.xuexiyuan.flinkstudy.sql;

import cn.xuexiyuan.flinkstudy.entity.Order;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Description: 将 DataStream 注册为 Table 和 View并进行 SQL 统计
 *
 * @Author 左龙龙
 * @Date 21-3-29
 * @Version 1.0
 **/
public class Demo01 {

    public static void main(String[] args) throws Exception{
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 1.source
        DataStreamSource<Order> orderA = env.fromCollection(Arrays.asList(
                new Order("1", "ELEM", 1, 40, System.currentTimeMillis()),
                new Order("2", "ELEM", 2, 22, System.currentTimeMillis()),
                new Order("3", "ELEM", 3, 40, System.currentTimeMillis()),
                new Order("4", "ELEM", 4, 33, System.currentTimeMillis()),
                new Order("5", "ELEM", 5, 40, System.currentTimeMillis())
        ));

        DataStreamSource<Order> orderB = env.fromCollection(Arrays.asList(
                new Order("1", "MTMW", 11, 40, System.currentTimeMillis()),
                new Order("2", "MTMW", 12, 22, System.currentTimeMillis()),
                new Order("3", "MTMW", 13, 40, System.currentTimeMillis()),
                new Order("4", "MTMW", 41, 33, System.currentTimeMillis()),
                new Order("5", "MTMW", 15, 40, System.currentTimeMillis())
        ));

        // 2.transformation
        Table tableA = tableEnv.fromDataStream(orderA, $("sheetno"), $("channel_keyword"), $("payment_money"), $("user_id"), $("order_time"));
        tableEnv.createTemporaryView("tableB", orderB, $("sheetno"), $("channel_keyword"), $("payment_money"), $("user_id"), $("order_time"));
        // 查询 tableA 中 payment_money > 1 和 tableB 中 payment_mondy > 3 的数据最后合并
        String sql = "select * from " + tableA + " where payment_money > 1 " +
                "union " +
                "select * from tableB where payment_money > 3";
        Table unionTable = tableEnv.sqlQuery(sql);
        unionTable.printSchema();

        // 将 Table 准换为 DataStream
        // toAppendStream -> 将计算后的数据 append 到结果 DataStream 中去
        sql = "select * from tableB";
        Table resultB = tableEnv.sqlQuery(sql);
        DataStream<Order> orderBDS = tableEnv.toAppendStream(resultB, Order.class);
        orderBDS.print("tableB: ");

        // toRetractStream -> 将计算后的新数据在 DataStream 原数据的基础上更新 true 或删除 false
        DataStream<Tuple2<Boolean, Order>> orderDS = tableEnv.toRetractStream(unionTable, Order.class);

        // 3.sink
        orderDS.print("union result: ");

        // 4.excute
        env.execute();

    }
}
